package com.ruoyi.mq.ntfy;

import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Service;
import org.springframework.web.reactive.function.client.WebClient;
import reactor.core.publisher.Flux;

import java.time.Duration;

@Service
@Slf4j
public class NtfySseListener implements ApplicationRunner {

    // 替换成你的NTFY服务器地址和主题
    private static final String NTFY_SSE_URL = "http://ntfy.yutang1314.fun:9000/test_ntfy/sse";

    private final WebClient webClient;

    public NtfySseListener(WebClient.Builder webClientBuilder) {
        this.webClient = webClientBuilder.build();
    }

    @Override
    public void run(ApplicationArguments args) {
        listenToNtfy();
    }

    public void listenToNtfy() {
        Flux<String> eventStream = webClient.get()
                .uri(NTFY_SSE_URL)
                .retrieve()
                .bodyToFlux(String.class)
                .timeout(Duration.ofHours(1)) // 设置超时时间，可根据需要调整
                .retry(3); // 失败重试3次

        eventStream.subscribe(
                message -> {
                    // 处理接收到的每一条消息
//                    log.info("收到NTFY消息: {}", message);
                    processMessage(message);
                },
                error -> {
                    // 处理错误（如连接断开）
                    log.error("监听NTFY主题时发生错误: {}", error.getMessage());
                    // 这里可以添加重连逻辑
                },
                () -> log.info("监听NTFY主题结束") // 流完成时（通常不会发生）
        );
    }

    private void processMessage(String message) {
        // 在这里实现你的业务逻辑
        // 例如：解析JSON消息（如果消息是JSON格式）、存入数据库、触发事件等
        if ("Hi".equals(message.trim())) {
            log.info("收到问候消息");
        }
        // ... 其他处理
    }
}